package com.atguigu.gmall.realtime.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{OrderDetail, OrderInfo, OrderWide}
import com.atguigu.gmall.realtime.utils.{ESUtil, KafkaUtil, OffsetUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.time.{LocalDate, Period}
import java.util
import scala.collection.mutable.ListBuffer

/**
 * @author caodan
 * @date 2022-09-06 16:06
 * @version 1.0
 */
object BaseOrderApp {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new
        SparkConf().setMaster("local[1]").setAppName("base-order-app")


    val ssc = new StreamingContext(sparkConf, Seconds(5))


    val orderInfoTopicName: String = "DWD_ORDER_INFO_I"
    val orderdetailTopicName: String = "DWD_ORDER_DETAIL_I"

    val groupId: String = "DWS_ORDER_GROUP"

    //从redis中获取偏移量
    val orderInfoOffsets: Map[TopicPartition, Long] = OffsetUtil.getOffset(orderInfoTopicName, groupId)

    val orderDetailOffsets: Map[TopicPartition, Long] = OffsetUtil.getOffset(orderdetailTopicName, groupId)

    // orderInfo
    var orderInfoKafkaDStream: DStream[ConsumerRecord[String, String]] = null
    if (orderInfoOffsets != null && orderInfoOffsets.nonEmpty) {
      orderInfoKafkaDStream = KafkaUtil.getKafkaDstream(orderInfoTopicName, ssc, groupId, orderInfoOffsets)
    } else {
      orderInfoKafkaDStream = KafkaUtil.getKafkaDstream(orderInfoTopicName, ssc, groupId)
    }
    // orderDetail
    var orderDetailKafkaDStream: DStream[ConsumerRecord[String, String]] = null
    if (orderDetailOffsets != null && orderDetailOffsets.nonEmpty) {
      orderDetailKafkaDStream = KafkaUtil.getKafkaDstream(orderdetailTopicName, ssc, groupId, orderDetailOffsets)
    } else {
      orderDetailKafkaDStream = KafkaUtil.getKafkaDstream(orderdetailTopicName, ssc, groupId)
    }

    //获取保存redis偏移量
    var orderInfoOffsetRanges: Array[OffsetRange] = null
    var orderDetailOffsetRanges: Array[OffsetRange] = null
    // 获取orderInfo的偏移量
    val orderInfoOffsetsDstream: DStream[ConsumerRecord[String, String]] = orderInfoKafkaDStream.transform(
      (rdd: RDD[ConsumerRecord[String, String]]) => {
        orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //获取orderDetali的偏移量
    val orderDetailOffsetsDStream: DStream[ConsumerRecord[String, String]] = orderDetailKafkaDStream.transform(
      (rdd: RDD[ConsumerRecord[String, String]]) => {
        orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //处理数据
    val orderInfoDStream: DStream[OrderInfo] = orderInfoOffsetsDstream.map(
      (consumerRecord: ConsumerRecord[String, String]) => {
        val value: String = consumerRecord.value()
        val orderInfo: OrderInfo = JSON.parseObject(value, classOf[OrderInfo])
        orderInfo
      }
    )

    val orderDetailDStream: DStream[OrderDetail] = orderDetailOffsetsDStream.map(
      (consumerRecord: ConsumerRecord[String, String]) => {
        val valueJson: String = consumerRecord.value()
        val orderDetail: OrderDetail = JSON.parseObject(valueJson, classOf[OrderDetail])
        orderDetail
      }
    )

    // 进行维度关联
    val orderInfoDimDStream: DStream[OrderInfo] = orderInfoDStream.mapPartitions(
      (iter: Iterator[OrderInfo]) => {
        val jedis: Jedis = RedisUtil.getJedisConn
        val orderInfoes: Iterator[OrderInfo] = iter.map(
          (orderInfo: OrderInfo) => {
            val userId: Long = orderInfo.user_id
            val provinceId: Long = orderInfo.province_id
            //拼接redis中的key
            val redisKeyByUserId = s"DIM:user_info:$userId"
            val redisKeybyProvinceId = s"DIM:base_province:$provinceId"
            val redisValueByUserIdJson: String = jedis.get(redisKeyByUserId)
            val redisValueByProIdJson: String = jedis.get(redisKeybyProvinceId)
            val userInfoJsonObj: JSONObject = JSON.parseObject(redisValueByUserIdJson)
            orderInfo.user_gender = userInfoJsonObj.getString("gender")
            val birthday: String = userInfoJsonObj.getString("birthday")
            //年龄计算
            val birthdayDate: LocalDate = LocalDate.parse(birthday)
            val nowDate: LocalDate = LocalDate.now()
            val period: Period = Period.between(birthdayDate, nowDate)
            val years: Int = period.getYears
            val age: Int = years
            orderInfo.user_age = age
            //地区维度属性
            val proIdJsonObj: JSONObject = JSON.parseObject(redisValueByProIdJson)
            orderInfo.province_name = proIdJsonObj.getString("name")
            orderInfo.province_iso_code = proIdJsonObj.getString("iso_code")
            orderInfo.province_3166_2_code = proIdJsonObj.getString("iso_3166_2")
            orderInfo.province_area_code = proIdJsonObj.getString("area_code")
            // 处理日期
            val createDt: String = orderInfo.create_time
            val createDrArr: Array[String] = createDt.split(" ")
            orderInfo.create_date = createDrArr(0)
            orderInfo.create_hour = createDrArr(0).split(":")(0)
            orderInfo
          }
        )
        RedisUtil.closeJedisConn(jedis)
        orderInfoes
      }
    )

    val orderInfoKVDStream: DStream[(Long, OrderInfo)] = orderInfoDimDStream.map((orderInfo: OrderInfo) => (orderInfo.id, orderInfo))

    val orderDetailKVDStream: DStream[(Long, OrderDetail)] = orderDetailDStream.map((orderDetail: OrderDetail) => (orderDetail.order_id, orderDetail))

    //双流join 操作
    val fullJoinStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoKVDStream.fullOuterJoin(orderDetailKVDStream)

    val orderWideDStream: DStream[OrderWide] = fullJoinStream.flatMap {
      case (_, (orderInfoOpt, orderDetailOpt)) => {
        val orderWideList: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
        val jedis: Jedis = RedisUtil.getJedisConn
        //1.主表在
        if (orderInfoOpt.isDefined) {
          val orderInfo: OrderInfo = orderInfoOpt.get
          //1.1 判断从表是否存在
          if (orderDetailOpt.isDefined) {
            //如果从表在， 将主和从合并
            val orderDetail: OrderDetail = orderDetailOpt.get
            val orderWide = new OrderWide(orderInfo, orderDetail)
            orderWideList.append(orderWide)
          }

          //1.2 主表写缓存
          //type: String
          //key : ORDER_JOIN:ORDER_INFO:[ID]
          //value : orderInfoJson
          //写入API: set
          //读取API: get
          //过期时间: 小时~天  24小时
          val orderInfoKey = s"ORDER_JOIN:ORDER_INFO:${orderInfo.id}"
          val orderInfoJson: String = JSON.toJSONString(orderInfo, new SerializeConfig(true))
          jedis.setex(orderInfoKey, 3600 * 24, orderInfoJson)

          //1.3 主表读缓存
          // type : set
          // key : ORDER_JOIN:ORDER_DETAIL:[ORDERID]
          // value : orderDetailJson
          //写入API: sadd
          //读取API: smembers
          //过期时间： 小时~天 24小时
          val orderDetailKey = s"ORDER_JOIN:ORDER_DETAIL:${orderInfo.id}"
          val orderDetaiJsonSet: util.Set[String] = jedis.smembers(orderDetailKey)
          import scala.collection.JavaConverters._
          if (orderDetaiJsonSet != null && orderDetaiJsonSet.size() > 0) {
            for (orderDetailJson <- orderDetaiJsonSet.asScala) {
              val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
              val orderWide = new OrderWide(orderInfo, orderDetail)
              orderWideList.append(orderWide)
            }
          }
        } else {
          //2. 主表不在，从表在
          val orderDetail: OrderDetail = orderDetailOpt.get
          //2.1 从表读缓存
          val orderInfoKey = s"ORDER_JOIN:ORDER_INFO:${orderDetail.order_id}"
          val orderInfoJson: String = jedis.get(orderInfoKey)
          if (orderInfoJson != null && orderInfoJson.nonEmpty) {
            val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
            val orderWide = new OrderWide(orderInfo, orderDetail)
            orderWideList.append(orderWide)
          }

          //2.2 从表写缓存
          val orderDetailKey = s"ORDER_JOIN:ORDER_DETAIL:${orderDetail.order_id}"
          val orderDetailJson: String = JSON.toJSONString(orderDetail, new SerializeConfig(true))
          jedis.sadd(orderDetailKey, orderDetailJson)
          jedis.expire(orderDetailKey, 3600 * 24)
        }

        jedis.close()
        orderWideList
      }
    }

    //写入es
    orderWideDStream.foreachRDD(
      (rdd: RDD[OrderWide]) => {
        //driver
        rdd.foreachPartition(
          (orderWideIter: Iterator[OrderWide]) => {
            //executor
            val orderWideList: List[(String, OrderWide)] =
              orderWideIter.toList.map((orderWide: OrderWide) => (orderWide.detail_id.toString , orderWide))
            if(orderWideList.nonEmpty ){
              val orderWideT: (String , OrderWide) = orderWideList.head
              val dt: String = orderWideT._2.create_date
              val indexName = s"gmall_order_wide_$dt"
              ESUtil.bulkSaveIdempotent(orderWideList,indexName)
            }
          }
        )
        //提交偏移量
        OffsetUtil.saveOffset(orderInfoTopicName,groupId , orderInfoOffsetRanges)
        OffsetUtil.saveOffset(orderdetailTopicName,groupId,orderDetailOffsetRanges)
      }
    )
    ssc.start()

    ssc.awaitTermination()

  }
}
